package com.baymax.study.tuLing.nio;

import lombok.SneakyThrows;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author baymax
 * @since 2025-06-27
 */
public class NioNonBlockDemo4_Thread {
    public static final int PORT = 60001;

    static class ServerBoss {
        public static void main(String[] args) throws Exception {
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);
            ssc.bind(new InetSocketAddress(PORT));

            Selector boss = Selector.open();
            ssc.register(boss, SelectionKey.OP_ACCEPT);
            System.out.println("服务器启动成功");

            ServerWorker worker = new ServerWorker("worker-1");
            while (true) {
                boss.select();
                Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    if (key.isAcceptable()) {
                        SocketChannel sc = ssc.accept();
                        sc.configureBlocking(false);
                        worker.register(sc); // boss线程处理accept事件，worker线程处理read事件
                    }
                }
            }
        }
    }

    static class ServerWorker implements Runnable {
        private Selector selector;
        private Thread thread;
        private String name;
        private volatile boolean start = false;

        private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();

        public ServerWorker(String name) {
            this.name = name;
        }

        public void register(SocketChannel sc) throws Exception {
            if (!start) {
                thread = new Thread(this, name);
                selector = Selector.open();
                thread.start();
                start = true;
            }
            queue.put(() -> {
                try {
                    sc.register(selector, SelectionKey.OP_READ);
                } catch (ClosedChannelException e) {
                    throw new RuntimeException(e);
                }
            });
            selector.wakeup(); // 唤醒selector
        }

        @SneakyThrows
        @Override
        public void run() {
            while (true) {
                selector.select();
                Runnable task = queue.poll();
                if (task != null) {
                    task.run(); // 执行注册任务 sc.register(selector, SelectionKey.OP_READ);
                }
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    if (key.isReadable()) {
                        SocketChannel sc = (SocketChannel) key.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        int bufferLth = sc.read(buffer);
                        if (bufferLth == -1) {
                            sc.close();
                            key.cancel();
                        } else {
                            buffer.flip();
                            System.out.println(Charset.defaultCharset().decode(buffer));
                        }
                    }
                }
            }
        }
    }

    static class Client {
        public static void main(String[] args) throws Exception {
            SocketChannel sc = SocketChannel.open();
            sc.connect(new InetSocketAddress("127.0.0.1", PORT));
            System.out.println("连接成功");

            TimeUnit.SECONDS.sleep(1);
            sc.write(ByteBuffer.wrap("hello client".getBytes()));
            System.out.println("数据发送完毕");
            sc.close();
        }
    }
}
